#include "panda.h"
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <errno.h>
#include <malloc.h>
#include <string.h>
#include <pthread.h>
#include "Defines.h"
#include "Facade.h"
#include "Threads.h"
#include "Debug.h"
/*
 * Initializes a connection queue.
 */
void Threads :: cq_init(CQ *cq) 
{
    pthread_mutex_init(&cq->lock, NULL);
    pthread_cond_init(&cq->cond, NULL);
    cq->head = NULL;
    cq->tail = NULL;
}

/*
 * Waits for work on a connection queue.
 */
CQ_ITEM *Threads :: cq_pop(CQ *cq) 
{
    CQ_ITEM *item;
    pthread_mutex_lock(&cq->lock);
    while (NULL == cq->head)
        pthread_cond_wait(&cq->cond, &cq->lock);
    item = cq->head;
    cq->head = item->next;
    if (NULL == cq->head)
        cq->tail = NULL;
    pthread_mutex_unlock(&cq->lock);
    return item;
}

/*
 * Looks for an item on a connection queue, but doesn't block if there isn't
 * one.
 * Returns the item, or NULL if no item is available
 */
CQ_ITEM *Threads :: cq_peek(CQ *cq) 
{
    CQ_ITEM *item;
    pthread_mutex_lock(&cq->lock);
    item = cq->head;
    if (NULL != item) {
        cq->head = item->next;
        if (NULL == cq->head)
            cq->tail = NULL;
    }
    pthread_mutex_unlock(&cq->lock);
    return item;
}

/*
 * Adds an item to a connection queue.
 */
void Threads :: cq_push(CQ *cq, CQ_ITEM *item) 
{
    item->next = NULL;
    pthread_mutex_lock(&cq->lock);
    if (NULL == cq->tail)
        cq->head = item;
    else
        cq->tail->next = item;
    cq->tail = item;
    pthread_cond_signal(&cq->cond);
    pthread_mutex_unlock(&cq->lock);
}

/*
 * Returns a fresh connection queue item.
 */
CQ_ITEM *Threads :: cqi_new() 
{
    CQ_ITEM *item = NULL;
    pthread_mutex_lock(&cqi_freelist_lock);
    if (cqi_freelist != NULL) {
        item = cqi_freelist;
        cqi_freelist = item->next;
    }
    pthread_mutex_unlock(&cqi_freelist_lock);

    if (NULL == item) {
        int i;
        /* Allocate a bunch of items at once to reduce fragmentation */
        item = (CQ_ITEM *)malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
        if (NULL == item)
            return NULL;

        /*
         * Link together all the new items except the first one
         * (which we'll return to the caller) for placement on
         * the freelist.
         */
        for (i = 2; i < ITEMS_PER_ALLOC; i++)
            item[i - 1].next = &item[i];
        pthread_mutex_lock(&cqi_freelist_lock);
        item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;
        cqi_freelist = &item[1];
        pthread_mutex_unlock(&cqi_freelist_lock);
    }

    return item;
}


/*
 * Frees a connection queue item (adds it to the freelist.)
 */
void Threads :: cqi_free(CQ_ITEM *item) 
{
    pthread_mutex_lock(&cqi_freelist_lock);
    item->next = cqi_freelist;
    cqi_freelist = item;
    pthread_mutex_unlock(&cqi_freelist_lock);
}


/*
 * Creates a worker thread.
 */
void Threads:: create_worker(void *(*func)(void *), void *arg)
 {
    pthread_t       thread;
    pthread_attr_t  attr;
    int             ret;
    pthread_attr_init(&attr);
    if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) {
        fprintf(stderr, "Can't create thread: %s\n",
                strerror(ret));
        exit(1);
    }
}


/*
 * Pulls a conn structure from the freelist, if one is available.
 */
conn * Threads::conn_from_freelist() 
{

	Facade *pR = Facade::Instance();
	conn *c;
	pthread_mutex_lock(&conn_lock);
	c = pR->GetNetWork()->do_conn_from_freelist();
	pthread_mutex_unlock(&conn_lock);
	return c;
}


/*
 * Adds a conn structure to the freelist.
 *
 * Returns 0 on success, 1 if the structure couldn't be added.
 */
bool Threads :: conn_add_to_freelist(conn *c) 
{
	bool result;
	Facade *pR = Facade::Instance();
	pthread_mutex_lock(&conn_lock);
	result = pR->GetNetWork()->do_conn_add_to_freelist(c);
	pthread_mutex_unlock(&conn_lock);
	return result;
}

/****************************** LIBEVENT THREADS *****************************/
/*
 * Set up a thread's information.
 */
void Threads :: setup_thread(LIBEVENT_THREAD *me)
 {
    if (! me->base) {
        me->base = event_init();
        if (! me->base) {
            fprintf(stderr, "Can't allocate event base\n");
            exit(1);
        }
    }
    if(me->thread_id == 0)
            me->thread_id = pthread_self();

    /* Listen for notifications from other threads */
    event_set(&me->notify_event, me->notify_receive_fd, EV_READ | EV_PERSIST, Thread_OnRead_Process, me);
    event_base_set(me->base, &me->notify_event);

    if (event_add(&me->notify_event, 0) == -1) {
        fprintf(stderr, "Can't monitor libevent notify pipe\n");
        exit(1);
    }

    cq_init(&me->new_conn_queue);
}


/*
 * Dispatches a new connection to another thread. This is only ever called
 * from the main thread, either during initialization (for UDP) or because
 * of an incoming connection.
 */
void Threads:: dispatch_conn_new(int sfd, int init_state, int event_flags, int read_buffer_size, int is_udp)
{
 Facade *pR = Facade::Instance();
    CQ_ITEM *item = cqi_new();
    int thread = (last_thread + 1) % pR->GetSettings()->num_threads;

    last_thread = thread;

    item->sfd = sfd;
    item->init_state = init_state;
    item->event_flags = event_flags;
    item->read_buffer_size = read_buffer_size;
    item->is_udp = is_udp;

    cq_push(&threads[thread].new_conn_queue, item);

    //MEMCACHED_CONN_DISPATCH(sfd, threads[thread].thread_id);
    if (write(threads[thread].notify_send_fd, "", 1) != 1) {
        perror("Writing to thread notify pipe");
    }
}

/*
 * Returns true if this is the thread that listens for new TCP connections.
 */
int Threads:: is_listen_thread() 
{
	return pthread_self() == threads[0].thread_id;
}

/*
 * Initializes the thread subsystem, creating various worker threads.
 * nthreads  Number of event handler threads to spawn
 * main_base Event base for main thread
 */
void Threads :: thread_init(int nthreads, struct event_base *main_base) 
{
	int         i;
	pthread_mutex_init(&conn_lock, NULL);
	pthread_mutex_init(&init_lock, NULL);
	pthread_cond_init(&init_cond, NULL);

	pthread_mutex_init(&cqi_freelist_lock, NULL);
	cqi_freelist = NULL;

	threads = (LIBEVENT_THREAD *) calloc(nthreads, sizeof(LIBEVENT_THREAD));
//	threads = (LIBEVENT_THREAD *) malloc(sizeof(LIBEVENT_THREAD) * nthreads);
	if (! threads) {
		perror("Can't allocate thread descriptors");
		exit(1);
	}

	threads[0].base = main_base;
	threads[0].thread_id = pthread_self();
	printf("%s: threads = %d\n", __func__, nthreads);
	for (i = 0; i < nthreads; i++) {
		int fds[2];
		if (pipe(fds)) {
			perror("Can't create notify pipe");
			exit(1);
		}

		threads[i].notify_receive_fd = fds[0];
		threads[i].notify_send_fd = fds[1];

		setup_thread(&threads[i]);
	}
DEBUG2("threads = ", nthreads);
	/* Create threads after we've done all the libevent setup. */
	for (i = 1; i < nthreads; i++) {
		create_worker(Thread_Worker_Start, &threads[i]);
	}
DEBUG1("create_worker! ");
	/* Wait for all the threads to set themselves up before returning. */
	pthread_mutex_lock(&init_lock);
	init_count++; /* main thread */
	while (init_count < nthreads) {
		pthread_cond_wait(&init_cond, &init_lock);
	}
	pthread_mutex_unlock(&init_lock);
}

int Threads::InitCountIncrease()
{
    int ic = 0;
    pthread_mutex_lock(&init_lock);
    init_count++;
    ic = init_count;
    pthread_cond_signal(&init_cond);
    pthread_mutex_unlock(&init_lock);
    return ic;
}

void Threads :: conn_listen_accept(bool do_accept)
{
    Facade *pR = Facade::Instance();
    pthread_mutex_lock(&conn_lock);
    pR->GetNetWork()->conn_listen_accept(do_accept);
    pthread_mutex_unlock(&conn_lock);
}


